Fixed

The client should be closed?

Fixed

Why not call flush() method directly?

+1

Fixed

Will it break the behavior of

// We cannot group acks if the delay is 0 or when there are properties attached to it. Fortunately that's an
// uncommon condition since it's only used for the compaction subscription.

If the consumer is reconnecting but the ack has properties. We will also group the acks which is not expected?

@codelipenghui Yes. But this behavior is only added for the TwoPhaseCompactor to avoid latency when seeking, the client side has no way to attach properties. see

Currently, if acknowledge is called during reconnection, phaseTwoSeekThenLoop will fail.

From my perspective, we should also queue the ACK requests and flush them after connected. @poorbarcode

Agress with @BewareMyPower

I am trying to split this PR into the following.

• part-1: add a new component AcknowledgmentCache to support caching the acknowledgments which include the arg properties.
• part-2: split PersistentAcknowledgmentsGroupingTracker into two implementations:
  • cache and batch the acks
  • immediately ack
• part-3: fix the issue

Marked current PR as Draft

split PersistentAcknowledgmentsGroupingTracker into two implementations

It's reasonable. BTW, the C++ client is already implemented in such way that:

• AckGroupingEnabled: cache the acknowledgments and flush them due to the timeout or count
• AckGrouingDisabled: acknowledge immediately

https://github.com/apache/pulsar-client-cpp/blob/72b7311aeef32e28a28e926da686aaf948e8f948/lib/ConsumerImpl.cc#L201C1-L215C6

Though the naming is not good (but consistent with other impl classes in the library)

Method invocations like f(args..., true) and f(args..., false) are hard to read unless you jump to the implementation of f and see what the boolean parameter means.

You should add a new method like queueIndividualAck and call it directly. For example,

    private CompletableFuture<Void> queueIndividualAck(MessageIdAdv messageId) {
        Optional<Lock> readLock = acquireReadLock();
        try {
            doIndividualAckAsync(messageId);
            return readLock.map(__ -> currentIndividualAckFuture).orElse(CompletableFuture.completedFuture(null));
        } finally {
            readLock.ifPresent(Lock::unlock);
            if (pendingIndividualAcks.size() >= maxAckGroupSize) {
                flush();
            }
        }
    }

    private CompletableFuture<Void> doIndividualAck(MessageIdAdv messageId, Map<String, Long> properties) {
        if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
            // We cannot group acks if the delay is 0 or when there are properties attached to it. Fortunately that's an
            // uncommon condition since it's only used for the compaction subscription.
            return doImmediateAck(messageId, AckType.Individual, properties, null);
        } else {
            return queueIndividualAck(messageId);
        }
    }

Then you don't have to add false argument to all existing doIndividualAck calls. And in doImmediateAck, you can call queueIndividualAck directly.

    private CompletableFuture<Void> doImmediateAck(MessageIdAdv msgId, AckType ackType, Map<String, Long> properties,
                                                   BitSetRecyclable bitSet) {
        ClientCnx cnx = consumer.getClientCnx();

        if (cnx == null && consumer.getState() == HandlerState.State.Connecting) {
            if (ackType == AckType.Cumulative) {
                return queueCumulativeAck(msgId);
            } else {
                return queueIndividualAck(msgId);
            }
        }

Sure, I will change the implementation of this PR as #21928 (comment)